package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.Aggregator;
import org.apache.kafka.streams.kstream.Initializer;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
@Slf4j
public class HotArticleStreamHandle {
    @Bean
    public KStream<String, String> KStream(StreamsBuilder streamsBuilder) {
        //接收消息
        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
        //聚合流式处理
        stream.map((key, value) -> {
                    UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
                    //重置消息的key:文章id，value: likes:1
                    return new KeyValue<>(mess.getArticleId().toString(), mess.getType().name() + ":" + mess.getAdd());
                }).groupBy((key, value) -> value)
                //时间窗口
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                //仔细完成聚合的操作
                .aggregate(new Initializer<String>() {
                               @Override
                               //初始方法 返回值是消息的value
                               public String apply() {
                                   return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                               }
                               //真正的聚合操作 返回值是消息的value
                           }, new Aggregator<String, String, String>() {
                               @Override
                               public String apply(String key, String value, String aggValue) {
                                   if (StringUtils.isBlank(value)) {
                                       return aggValue;
                                   }
                                   String[] aggAry = aggValue.split(",");
                                   int col = 0, com = 0, lik = 0, vie = 0;
                                   for (String agg : aggAry) {
                                       String[] split = agg.split(":");
                                       switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])) {
                                           /**
                                            * 获得初始值 也是时间窗口内计算之后的值
                                            */
                                           case COLLECTION:
                                               col = Integer.valueOf(split[1]);
                                               break;
                                           case COMMENT:
                                               com = Integer.valueOf(split[1]);
                                               break;
                                           case LIKES:
                                               lik = Integer.valueOf(split[1]);
                                               break;
                                           case VIEWS:
                                               vie = Integer.valueOf(split[1]);
                                               break;
                                       }
                                   }
                                   //mess.getArticleId().toString(), mess.getType().name() + ":" + mess.getAdd()
                                   /**
                                    * 累加操作
                                    */
                                   String[] valAry = value.split(":");
                                   switch (UpdateArticleMess.UpdateArticleType.valueOf(valAry[0])) {
                                       case COLLECTION:
                                           col += Integer.valueOf(valAry[1]);
                                           break;
                                       case COMMENT:
                                           com += Integer.valueOf(valAry[1]);
                                           break;
                                       case LIKES:
                                           lik += Integer.valueOf(valAry[1]);
                                           break;
                                       case VIEWS:
                                           vie += Integer.valueOf(valAry[1]);
                                           break;
                                   }
                                   String formatstr = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", col, com, lik, vie);
                                   System.out.println("文章id"+key);
                                   System.out.println("当前时间窗口内的消息处理结果:"+formatstr);
                                   return formatstr;
                               }
                           }, Materialized.as("hot-article-stream-count-001")
                ).toStream()
                .map((key,value)->
                {
                    return new KeyValue<>(key.key().toString(),formatObj(key.key().toString(),value));
                })
                //发送消息
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);
        //发送消息
        return stream;
    }
    //格式化消息的value数据
    public String formatObj(String articleId,String value){
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        mess.setArticleId(Long.valueOf(articleId));
        //COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
        String[] valAry = value.split(",");
        for (String val : valAry) {
            String[] split = val.split(":");
            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])) {
                case COLLECTION:
                    mess.setCollect(Integer.valueOf(split[1]));
                    break;
                case COMMENT:
                    mess.setComment(Integer.valueOf(split[1]));
                    break;
                case LIKES:
                    mess.setLike(Integer.valueOf(split[1]));
                    break;
                case VIEWS:
                    mess.setCollect(Integer.valueOf(split[1]));
                    break;
            }
        }
        log.info("聚合消息之后的结果为：{}",JSON.toJSONString(mess));
        return JSON.toJSONString(mess);

    }
}
